package org.example.demo;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Stream;

/**
 * @author Dache
 */
public abstract class AsyncSubscriber<T> implements Subscriber<T>,Runnable {
    // Signal 表示发布者和订阅者之间的异步协议
    private static interface Signal {
    }

    // 表示数据流发送完成，完成信号
    private enum OnComplete implements Signal {Instance;}

    // 表示发布者给订阅者的异常信号
    private static class OnError implements Signal {
        public final Throwable t;

        public OnError(final Throwable t) {
            this.t = t;
        }
    }

    // 表示下一个数据项的信号
    private static class OnNext<T> implements Signal {
        private final T next;

        public OnNext(final T next) {
            this.next = next;
        }
    }

    // 表示订阅者的订阅成功的信号
    private static class OnSubscribe implements Signal {
        public final Subscription subscription;

        public OnSubscribe(final Subscription subscription) {
            this.subscription = subscription;
        }
    }

    // 订阅单据，根据规范3.1 该引用是私有的
    private Subscription subscription;


    // 用于表示当前的订阅者是否处理完成
    private boolean done;

    // 根据规范的2.2条款，使用该线程池异步处理的各个信号
    private final Executor executor;

    /**
     * 仅有这一个构造器，只能被子类调用
     * 传递一个线程池即可
     *
     * @param executor 线程池对象
     */
    protected AsyncSubscriber(Executor executor) {
        if (executor == null) {
            throw null;
        }
        this.executor = executor;
    }

    /**
     * 幂等地标记当前订阅者已经完成处理，不再处理更多的元素
     * 因此，需要取消订阅票据（Subscription）
     */
    private final void done() {
        // 在此处。可以添加done，对订阅者的完成状态进行设置；
        // 虽然规范3.7规定Subscription.cancel()是幂等的，我们不需要这么做。
        // 当whenNext方法抛异常， 认为订阅者已经处理完成(不再接受更多元素)
        done = true;
        if (subscription != null) {
            // If we are bailing out before we got a `Subscription` there's little need for cancelling it.
            try {
                // 取消票据
                subscription.cancel();
            } catch (final Throwable t) {
                // 根据规范条款3.15 此处不能抛出异常。因此只能是记录下来。
                (new IllegalStateException(subscription + " violated the Reactive Streams rule 3.15 by throwing an " +
                        "exception from cancel.", t)).printStackTrace(System.err);
            }

        }


    }

    /**
     * This method is invoked when the OnNext signals arrive
     * Returns whether more elements are desired or not, and if no more elements are desired,
     * for convenience.
     * 当OnNext信号到达时调用此方法
     * 返回是否需要更多元素，如果不需要更多元素，
     * 为了方便。
     *
     * @param element
     * @return boolean
     */
    protected abstract boolean whenNext(final T elment);


    /**
     * This method is invoked when the OnComplete signal arrives
     * override this method to implement your own custom onComplete logic.
     * 此方法在OnComplete信号到达时调用
     * 重写此方法以实现您自己的自定义OnComplete逻辑。
     */
    protected void whenComplete() {
    }


    /**
     * This method is invoked if the OnError signal arrives
     * override this method to implement your own custom onError logic
     * 此方法在OnError信号到达时调用
     * 重写此方法以实现您自己的自定义OnError逻辑。
     *
     * @param error
     */
    protected void whenError(Throwable error) {
    }


    private final void handleOnSubscribe(final Subscription s) {
        if (s == null) {
            // Getting a null `Subscription` here is not valid so lets just ignore it
        } else if (subscription != null) {
            // If someone has made a mistake and added this Subscriber multiple times, let's handle it gracefully
            // 如果有人犯了错误并多次添加此订阅者，让我们优雅地处理它
            try {
                // Cancel the additional subscription to follow rule 2.5 取消额外订阅以遵循规则2.5
                s.cancel();
            } catch (final Throwable throwable) {
                //Subscription.cancel is not allowed to throw an exception,according to rule 3.15
                (new IllegalStateException(s + " violated the Reactive " +
                        "Streams rule 3.15 by throwing an exception from cancel.", throwable)).printStackTrace(System.err);
            }
        } else {
            // We have to assign it locally before we use it, if we want to be a synchronous `Subscriber`
            // 如果我们想成为同步的“订阅者”，我们必须在使用它之前在本地分配它
            //Because according to rule 3.10, the Subscription is allowed to call `onNext` synchronously from within `request
            // 因为根据规则3.10，允许订阅从“request”中同步调用“onNext”
            subscription = s;
            try {
                // If we want elements, according to rule 2.1 we need to call `request`
                // 如果我们想要元素，根据规则2.1，我们需要调用“request”
                //  And, according to rule 3.2 we are allowed to call this synchronously from within the `onSubscribe` method
                // 而且，根据规则3.2，我们可以从'onSubscribe'方法中同步调用它
                s.request(1);
                //  Our Subscriber is unbuffered and modest, it requests one element at a time
                // 我们的订户是无缓冲和谦虚的，它一次请求一个元素
            } catch (final Throwable t) {
                // Subscription.request is not allowed to throw according to rule 3.16
                // 根据规则3.16不允许抛出Subscript. request
                (new IllegalStateException(s + " violated the Reactive Streams rule 3.16 by throwing an exception from " +
                        "request.", t)).printStackTrace(System.err);
            }
        }
    }

    private final void handleOnNext(final T elemet) {
        if (!done) {
            if (subscription == null) {
                // Technically this check is not needed, since we are expecting Publishers to conform to the spec  Check for spec violation of 2.1 and 1.09
                //从技术上讲，不需要此检查，因为我们希望发布商符合规范检查2.1和1.09的规范违规
                (new IllegalStateException("Someone violated the Reactive Streams rule 1.09 and 2.1 by signalling OnNext" +
                        " before `Subscription.request`. (no Subscription)")).printStackTrace(System.err);
            }
        } else {
            try {
                if (whenNext(elemet)) {
                    try {
                        subscription.request(1);
                        // Our Subscriber is unbuffered and modest, it requests one element at a time
                        // 我们的订阅者是无缓冲和适度的，它一次请求一个元素
                    } catch (final Throwable t) {
                        // Subscription.request is not allowed to throw according to rule 3.16
                        (new IllegalStateException(subscription + " violated the Reactive Streams rule 3.16 by throwing" +
                                " an exception from request.", t)).printStackTrace(System.err);
                    }
                } else {
                    done();// This is legal according to rule 2.6
                }
            } catch (final Throwable t) {
                done();
                try {
                    onError(t);
                } catch (final Throwable t2) {
                    //Subscriber.onError is not allowed to throw an exception, according to rule 2.13
                    (new IllegalStateException(this + " violated the Reactive Streams rule 2.13 by throwing an exception" +
                            " from onError.", t2)).printStackTrace(System.err);
                }
            }
        }
    }

    /**
     * Here it is important that we do not violate 2.2 and 2.3 by calling
     * methods on the `Subscription` or `Publisher`
     * 在这里，重要的是我们不要通过调用“订阅”或“发布者”上的方法来违反2.2和2.3
     */
    private void handleOnComplete() {
        if (subscription == null) {
            // Technically this check is not needed, since we are expecting Publishers to conform to the spec
            // Publisher is not allowed to signal onComplete before onSubscribe according to rule 1.09
            //从技术上讲，这个检查是不需要的，因为我们希望发布者符合规范，根据规则1.09，发布者不允许在onSubscribe之前发出onComplete信号
            (new IllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onComplete prior " +
                    "to onSubscribe.")).printStackTrace(System.err);
        } else {
            // Obey rule 2.4
            done = true;
            whenComplete();
        }
    }


    /**
     * Here it is important that we do not violate 2.2 and 2.3 by
     * calling methods on the `Subscription` or `Publisher`
     *
     * @param error
     */
    private void handleOnError(final Throwable error) {
        if (subscription == null) {
            // Technically this check is not needed, since we are expecting Publishers to conform to the spec
            // Publisher is not allowed to signal onComplete before onSubscribe according to rule 1.09
            //从技术上讲，这个检查是不需要的，因为我们希望发布者符合规范，根据规则1.09，发布者不允许在onSubscribe之前发出onComplete信号
            (new IllegalStateException("Publisher violated the Reactive Streams rule 1.09 signalling onComplete prior " +
                    "to onSubscribe.")).printStackTrace(System.err);
        } else {
            // Obey rule 2.4
            done = true;
            whenError(error);
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public void run() {
        // 跟上次线程执行建立happens-before关系，防止多个线程并发执行
        /**
         * 在 Java 中，happens-before 是一种偏序关系（partial ordering），它定义了两个操作之间的可见性与顺序性约束：
         * 如果操作 A happens-before 操作 B，那么 A 的结果对 B 是可见的。
         * 并且，JVM 必须保证 A 的执行顺序在 B 之前。
         */
        if (on.get()) {
            try {
                // 从入站队列取出信号
                final Signal s = inboundSignals.poll();
                // 根据规范条款2.8，如果当前订阅者已完成，就不需要处理了
                if (!done) {
                    // 根据信号类型调用对应的方法来处理
                    if (s instanceof OnNext<?>) {
                        handleOnNext(((OnNext<T>) s).next);
                    } else if (s instanceof OnSubscribe) {
                        handleOnSubscribe(((OnSubscribe) s).subscription);
                    }
                    // 根据规范2.10，必须处理onError信号，不管有没有调用过 Subscription.request(long n)方法
                    else if (s instanceof OnError) {
                        handleOnError(((OnError) s).t);
                    }
                    // 根据规范2.9，必须处理onComplete信号，不管有没有调用过 Subscription.request(long n)方法
                    else if (s == OnComplete.Instance) {
                        handleOnComplete();
                    }
                }
            } finally {
                // 保持happens-before关系，然后开始下一个线程调度执行
                on.set(false);
                // 如果入站信号不是空的，调度线程处理入站信号
                if (!inboundSignals.isEmpty()) {
                    // 调度处理入站信号
                    tryScheduleToExecute();
                }
            }
        }
    }

    /**
     * We implement the OnX methods on `Subscriber` to send Signals that we
     * will process asycnhronously, but only one at a time
     * 我们在“订阅者”上实现OnX方法来发送我们将不同步处理，但一次只能处理一个
     *
     * @param subscription
     */
    @Override
    public void onSubscribe(Subscription s) {
        // As per rule 2.13, we need to throw a ‘java.lang.NullPointerException` if the `Subscription` is `null`
        if (s == null) {
            throw null;
        }
        signal(new OnSubscribe(s));
    }

    @Override
    public void onNext(final T element) {
        // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `element` is `null`
        if (element == null) {
            throw null;
        }
        signal(new OnNext<T>(element));
    }

    @Override
    public void onError(Throwable t) {
        // As per rule 2.13, we need to throw a `java.lang.NullPointerException` if the `Throwable` is `null`
        if (t == null) {
            throw null;
        }
        signal(new OnError(t));
    }

    @Override
    public void onComplete() {
        signal(OnComplete.Instance);
    }

    private final ConcurrentLinkedQueue<Signal> inboundSignals = new ConcurrentLinkedQueue<Signal>();

    /**
     * 根据规范2.7和2.11，使用原子变量确保不会有多个订阅者线程并发执行。
     */
    private final AtomicBoolean on = new AtomicBoolean(false);

    /**
     * What `signal` does is that it sends signals to the `Subscription`
     * asynchronously
     * 该方法异步地向订阅票据发送信号
     */
    private void signal(final Signal signal) {
        // 信号入站，线程池进行调度
        // 不需要检查是否为null，因为已经实例化了
        if (inboundSignals.offer(signal)) {
            // 线程调度处理
            tryScheduleToExecute();
        }
    }

    /**
     * 确保订阅者一次仅在一个线程执行
     * 调度执行
     */
    private final void tryScheduleToExecute() {
        // 使用CAS原子性地修改变量on的值改为true。
        if (on.compareAndSet(false, true)) {
            try {
                executor.execute(this);
            } catch (Throwable t) {
                // 根据规范条款2.13，如果不能执行线程池的提交方法，需要优雅的退出
                if (!done) {
                    try {
                        // 由于错误不可恢复，因此取消订阅票据
                        done();
                    } finally {
                        // 不再需要处理入站信号，清空之
                        inboundSignals.clear();
                        // 由于票据已经取消，但是此处依然让订阅者处于可调度的状态，以防在清空入站信号之后又有信号发送过来、
                        // 因为信号的发送是异步的
                        on.set(false);
                    }
                }

            }
        }

    }
}
